import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;

@Slf4j
public class HBaseMain {
    private static Connection connection = null;
    private static String ZK_ADDRESS = "127.0.0.1";
    private static String ZK_HOST = "2181";
    private static String HBASE_MASTER = "127.0.0.1:60000";

    public static void main(String[] args) throws Exception {
        if (args.length == 3) {
            ZK_ADDRESS = args[0];
            ZK_HOST = args[1];
            HBASE_MASTER = args[2];
        }
        // 建立连接
        Admin admin = getAdmin();
        try {
            //创建namespace
            String namespace = "chenxi";
            createNameSpace(admin, namespace);
            //建表
            TableName tableName = TableName.valueOf(String.format("%s:student", namespace));
            Set<String> colFamily = new HashSet<>();
            colFamily.add("info");
            colFamily.add("score");
            createTable(admin, tableName, colFamily);
            //塞数据
            List<StudentData> studentData = getNeedInsertData();
            batchInsert(studentData, tableName);

            //获取数据
            studentData.stream().map(v -> v.getName()).forEach(v -> {
                getData(tableName, v);
            });

            //删除数据
            studentData.stream().map(v -> v.getName()).forEach(v -> {
                deleteData(tableName, v);
            });
            //删除表
            removeTable(admin, tableName);

        } catch (Exception e) {
            log.error("HBaseMain error", e);
        } finally {
            if (admin != null) {
                admin.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }

    private static Connection getConnection() {
        if (connection != null) {
            return connection;
        }
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", ZK_ADDRESS);
        configuration.set("hbase.zookeeper.property.clientPort", ZK_HOST);
        configuration.set("hbase.master", HBASE_MASTER);
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            log.error("connect fail", e);
        }
        return connection;
    }

    private static Admin getAdmin() throws IOException {
        Admin admin = getConnection().getAdmin();
        return admin;
    }

    private static void createNameSpace(Admin admin, String nameSpace) throws IOException {
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
        try {
            admin.createNamespace(namespaceDescriptor);
        } catch (NamespaceExistException e) {
            log.warn("namespace already exists");
            return;
        }
        log.info("namespace create success");
    }

    private static List<StudentData> getNeedInsertData() {
        List<StudentData> studentData = new ArrayList<>();
        studentData.add(StudentData.builder().name("Tom").studentId("20210000000001").classId("1").understanding("75").programming("82").build());
        studentData.add(StudentData.builder().name("chenxi").studentId("G20220735020169").build());
        return studentData;
    }

    private static void batchInsert(List<StudentData> studentData, TableName tableName) {
        for (StudentData data : studentData) {
            List<ColumData> columData = new ArrayList<>();
            if (data.getStudentId() != null) {
                columData.add(ColumData.builder().family("info").qualifier("student_id").value(data.getStudentId()).build());
            }
            if (data.getClassId() != null) {
                columData.add(ColumData.builder().family("info").qualifier("class").value(data.getClassId()).build());
            }
            if (data.getUnderstanding() != null) {
                columData.add(ColumData.builder().family("score").qualifier("understanding").value(data.getUnderstanding()).build());
            }
            if (data.getProgramming() != null) {
                columData.add(ColumData.builder().family("score").qualifier("programming").value(data.getProgramming()).build());
            }
            putData(tableName, data.getName(), columData);
        }
    }

    /**
     * 建表
     *
     * @param admin
     * @param tableName
     * @param colFamily
     * @throws IOException
     */
    private static void createTable(Admin admin, TableName tableName, Set<String> colFamily) throws IOException {
        // 建表
        if (admin.tableExists(tableName)) {
            log.warn("Table already exists");
        } else {
            List<ColumnFamilyDescriptor> columnFamily = new ArrayList<>();
            if (!CollectionUtils.isEmpty(colFamily)) {
                colFamily.stream().forEach(v -> {
                    ColumnFamilyDescriptor colFamilyDescriptor = ColumnFamilyDescriptorBuilder.of(v);
                    columnFamily.add(colFamilyDescriptor);
                });
            }

            TableDescriptor hTableDescriptor =
                    TableDescriptorBuilder.newBuilder(tableName).setColumnFamilies(columnFamily).build();
            admin.createTable(hTableDescriptor);
            log.info("Table create successful");
        }
    }


    private static void putData(TableName tableName, String rowKey, List<ColumData> data) {
        if (CollectionUtils.isEmpty(data)) {
            return;
        }
        List<Put> puts = new ArrayList<>();
        // 插入数据
        data.stream().forEach(v -> {
            Put put = new Put(Bytes.toBytes(rowKey)); // row key
            put.addColumn(Bytes.toBytes(v.getFamily()), v.getQualifier() == null ? null :
                    Bytes.toBytes(v.getQualifier()), Bytes.toBytes(v.getValue())); //
            puts.add(put);
        });
        // col1
        try {
            getConnection().getTable(tableName).put(puts);
        } catch (IOException e) {
            log.info("Data insert fail", e);
        }
        log.info("Data insert success");
    }

    private static void getData(TableName tableName, String rowKey) {
        // 查看数据
        Get get = new Get(Bytes.toBytes(rowKey));
        if (!get.isCheckExistenceOnly()) {
            Result result = null;
            try {
                result = getConnection().getTable(tableName).get(get);
            } catch (IOException e) {
                log.error("getData error", e);
                return;
            }
            List<ColumData>columData=new ArrayList<>();
            for (Cell cell : result.rawCells()) {
                String colFamilyName = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(),
                        cell.getFamilyLength());
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
                        cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                ColumData colum=ColumData.builder().family(colFamilyName).qualifier(colName).value(value).build();
                columData.add(colum);
            }
            if(!CollectionUtils.isEmpty(columData)){
            StringBuffer stringBuffer=new StringBuffer(String.format("Data get success,Id:%s",rowKey));
            for (ColumData v:columData) {
                stringBuffer.append(String.format(",%s.%s:%s",v.getFamily(),v.getQualifier(),v.getValue()));
            }
            log.info(stringBuffer.toString());
            }
        }
    }

    private static void deleteData(TableName tableName, String rowKey) {
        // 删除数据
        Delete delete = new Delete(Bytes.toBytes(rowKey));      // 指定rowKey
        try {
            getConnection().getTable(tableName).delete(delete);
        } catch (IOException e) {
            log.info("Delete fail", e);
        }
        log.info("Delete Success");
    }

    private static void removeTable(Admin admin, TableName tableName) throws IOException {
        // 删除表
        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            log.info("Table Delete Successful");
        } else {
            log.info("Table does not exist!");
        }
    }
}
